#include "mqsimpleconsumer.h"
#include <QDebug>
#include <QMutexLocker>
#include <QCoreApplication>

MQSimpleConsumer::MQSimpleConsumer(std::shared_ptr<MQClient> client, QObject* parent)
    : QObject(parent)
    , m_client(client)
    , m_workerThread(nullptr)
    , m_pollTimer(nullptr)
    , m_consuming(false)
    , m_lastConnectionStatus(false)
{
    // 定时器将在startConsuming中动态创建
}

MQSimpleConsumer::~MQSimpleConsumer()
{
    stopConsuming(true);
    // 定时器会在工作线程结束时自动删除
}

bool MQSimpleConsumer::startConsuming(const QString& queueName, 
                                     std::shared_ptr<MQMessageHandler> handler,
                                     int pollInterval)
{
    if (m_consuming) {
        qWarning() << "MQSimpleConsumer::startConsuming() - Already consuming";
        return false;
    }
    
    if (!m_client) {
        qCritical() << "MQSimpleConsumer::startConsuming() - MQ client is null";
        return false;
    }
    
    if (!handler) {
        qCritical() << "MQSimpleConsumer::startConsuming() - Message handler is null";
        return false;
    }
    
    if (queueName.isEmpty()) {
        qCritical() << "MQSimpleConsumer::startConsuming() - Queue name is empty";
        return false;
    }
    
    // 设置参数
    m_queueName = queueName;
    m_handler = handler;
    m_consuming = true;
    
    // 重置统计信息
    {
        QMutexLocker locker(&m_statsMutex);
        m_stats = SimpleStats();
    }
    
    // 创建工作线程
    m_workerThread = new QThread();
    
    // 创建新的定时器在工作线程中
    QTimer* workerTimer = new QTimer();
    workerTimer->setInterval(pollInterval);
    workerTimer->setSingleShot(false);
    
    // 将定时器移动到工作线程
    workerTimer->moveToThread(m_workerThread);
    
    // 连接信号槽（跨线程连接）
    connect(workerTimer, &QTimer::timeout, this, &MQSimpleConsumer::pollAndProcess, Qt::QueuedConnection);
    connect(m_workerThread, &QThread::started, workerTimer, QOverload<>::of(&QTimer::start), Qt::QueuedConnection);
    connect(m_workerThread, &QThread::finished, workerTimer, &QTimer::deleteLater);
    
    // 保存工作线程定时器引用
    m_pollTimer = workerTimer;
    
    // 启动工作线程
    m_workerThread->start();
    
    qDebug() << "MQSimpleConsumer::startConsuming() - Started consuming queue:" 
             << queueName << "with interval:" << pollInterval << "ms";
    
    return true;
}

void MQSimpleConsumer::stopConsuming(bool waitForFinished)
{
    if (!m_consuming) {
        return;
    }
    
    qDebug() << "MQSimpleConsumer::stopConsuming() - Stopping consumer";
    
    m_consuming = false;
    
    // 停止并清理工作线程
    if (m_workerThread) {
        // 请求线程退出
        m_workerThread->quit();
        
        if (waitForFinished) {
            // 等待线程完成，最多等待5秒
            if (!m_workerThread->wait(5000)) {
                qWarning() << "MQSimpleConsumer::stopConsuming() - Thread did not finish within timeout, forcing termination";
                m_workerThread->terminate();
                m_workerThread->wait(1000);
            }
        }
        
        // 清理线程对象
        m_workerThread->deleteLater();
        m_workerThread = nullptr;
    }
    
    // 清理定时器引用（实际对象会在线程结束时自动删除）
    m_pollTimer = nullptr;
    
    qDebug() << "MQSimpleConsumer::stopConsuming() - Consumer stopped";
}

SimpleStats MQSimpleConsumer::getStats() const
{
    QMutexLocker locker(&m_statsMutex);
    return m_stats;
}

bool MQSimpleConsumer::isConsuming() const
{
    return m_consuming;
}

void MQSimpleConsumer::setPollInterval(int interval)
{
    if (interval < 10) {
        qWarning() << "MQSimpleConsumer::setPollInterval() - Interval too small, setting to 10ms";
        interval = 10;
    }
    
    // 只有在消费过程中且定时器存在时才设置间隔
    if (m_consuming && m_pollTimer) {
        QMetaObject::invokeMethod(m_pollTimer, "setInterval", Qt::QueuedConnection, Q_ARG(int, interval));
        qDebug() << "MQSimpleConsumer::setPollInterval() - Set to:" << interval << "ms";
    } else {
        qWarning() << "MQSimpleConsumer::setPollInterval() - Consumer not running or timer not available";
    }
}

void MQSimpleConsumer::pollAndProcess()
{
    if (!m_consuming || !m_client || m_queueName.isEmpty()) {
        return;
    }
    
    try {
        // 检查连接状态
        bool currentConnectionStatus = false;
        
        try {
            currentConnectionStatus = m_client->isConnected();
        } catch (const std::exception& e) {
            qCritical() << "MQSimpleConsumer::pollAndProcess() - Exception checking connection:" << e.what();
            currentConnectionStatus = false;
        } catch (...) {
            qCritical() << "MQSimpleConsumer::pollAndProcess() - Unknown exception checking connection";
            currentConnectionStatus = false;
        }
        
        // 连接状态变化处理
        if (currentConnectionStatus != m_lastConnectionStatus) {
            m_lastConnectionStatus = currentConnectionStatus;
            emit connectionStatusChanged(currentConnectionStatus);
            
            if (!currentConnectionStatus) {
                handleConnectionLost();
                return;
            } else {
                qDebug() << "MQSimpleConsumer::pollAndProcess() - Connection restored";
            }
        }
        
        // 如果未连接，跳过本次轮询
        if (!currentConnectionStatus) {
            return;
        }
        
        // 尝试接收消息（非阻塞）
        QByteArray message = m_client->consumeMessage(m_queueName, true); // 自动确认
        
        if (!message.isEmpty()) {
            qDebug() << "MQSimpleConsumer::pollAndProcess() - Received message from queue:" 
                     << m_queueName << "Size:" << message.size();
            

            // 发出消息接收信号 //GJ todo 后续可以高频消费
            emit messageReceived(m_queueName, message);
            
            // 直接同步处理消息
            bool success = processWithTimeout(message);
            
            // 更新统计信息
            updateStats(success);
            
            //GJ todo 后续可用
            // 发出处理完成信号
            emit messageProcessed(success);
        }
        
    } catch (const std::exception& e) {
        QString error = QString("Poll error: %1").arg(e.what());
        qCritical() << "MQSimpleConsumer::pollAndProcess() - Exception:" << error;
        handleError(error);
        
        // 连接可能已断开，强制检查连接状态
        m_lastConnectionStatus = false;
        
    } catch (...) {
        QString error = "Poll unknown error";
        qCritical() << "MQSimpleConsumer::pollAndProcess() - Unknown exception";
        handleError(error);
        
        // 连接可能已断开，强制检查连接状态
        m_lastConnectionStatus = false;
    }
}

void MQSimpleConsumer::handleConnectionLost()
{
    qWarning() << "MQSimpleConsumer::handleConnectionLost() - MQ connection lost";
    // 连接丢失时，暂停处理，等待下次轮询重新检查连接
}

void MQSimpleConsumer::updateStats(bool success)
{
    QMutexLocker locker(&m_statsMutex);
    
    m_stats.totalMessages++;
    
    if (success) {
        m_stats.successCount++;
    } else {
        m_stats.errorCount++;
    }
}

bool MQSimpleConsumer::processWithTimeout(const QByteArray& message, int timeoutMs)
{
    if (!m_handler) {
        qWarning() << "MQSimpleConsumer::processWithTimeout() - Handler is null";
        return false;
    }
    
    QElapsedTimer timer;
    timer.start();
    
    bool result = false;
    
    // 调用处理器
    result = m_handler->handleMessage("", "", message);
        
    qint64 elapsed = timer.elapsed();
    
    // 更新处理时间统计
    {
        QMutexLocker locker(&m_statsMutex);
        m_stats.lastProcessTime = elapsed;
    }
    
    if (elapsed > timeoutMs) {
        qWarning() << "MQSimpleConsumer::processWithTimeout() - Processing timeout:" 
                   << elapsed << "ms (limit:" << timeoutMs << "ms)";
    }
    
    qDebug() << "MQSimpleConsumer::processWithTimeout() - Processing completed in" 
             << elapsed << "ms, result:" << (result ? "success" : "failed");
    
    return result;
}

void MQSimpleConsumer::handleError(const QString& error)
{
    qCritical() << "MQSimpleConsumer::handleError() - Error:" << error;
    emit errorOccurred(error);
}
